consumer: sort fallback DML before flush - #5824
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
📝 WalkthroughWalkthroughThis PR removes the ChangesWatermark-ordered DML buffering
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XL
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
cmd/storage-consumer/consumer.go (1)
297-318: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winStorage consumer still drops fallback rows — inconsistent with the new sort-before-flush policy.
The Kafka and Pulsar writers now append unconditionally and rely on
ResolveIntoto order, but here a row withcommitTs < group.HighWatermarkis silently discarded unlessenableTableAcrossNodes. Given the group sorts before draining, the drop is no longer needed and causes exactly the sink inconsistency this PR targets.♻️ Align with the other consumers
- if commitTs >= group.HighWatermark { - group.AppendMessage(message) - log.Debug("DML event append to the group", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType)) - return - } - if enableTableAcrossNodes { - log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType)) - group.AppendMessage(message) - return - } - log.Warn("dml event commit ts fallback, ignore", - zap.Uint64("commitTs", commitTs), - zap.Any("highWatermark", group.HighWatermark), - zap.String("schema", schema), - zap.String("table", table), - ) + highWatermark := group.HighWatermark + group.AppendMessage(message) + if commitTs < highWatermark { + log.Warn("DML event commit ts fallback, append it and sort before flush", + zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", highWatermark), + zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), + zap.Stringer("eventType", message.RowType)) + }Note
enableTableAcrossNodesbecomes unused in this function; drop the parameter or keep it only as a log field.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/storage-consumer/consumer.go` around lines 297 - 318, Update the storage consumer’s DML handling around the group.HighWatermark checks to append every message unconditionally, including commitTs values below the high watermark, relying on the group’s sort-before-flush behavior. Remove the now-unnecessary enableTableAcrossNodes branch and parameter from the relevant function, or retain it only as a log field, and eliminate the fallback-drop path.
🧹 Nitpick comments (4)
cmd/kafka-consumer/writer.go (1)
619-643: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUnbounded
Warnper fallback row.During a broker disruption every re-consumed row hits one of these two
Warnpaths, which can flood the log at row rate. Consider rate-limiting (or aggregating per table/flush) and using typed fields (zap.Uint64forpartitionWatermark,zap.Stringerforprotocol) instead ofzap.Any.As per coding guidelines, "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/kafka-consumer/writer.go` around lines 619 - 643, Rate-limit or aggregate the fallback-row warnings in the relevant writer flow so broker disruptions cannot emit one Warn log per row, while preserving visibility per table or flush. Update the affected structured fields to use zap.Uint64 for partitionWatermark and zap.Stringer for protocol, following the logging guidance; keep the existing fallback behavior unchanged.Source: Coding guidelines
cmd/kafka-consumer/writer_test.go (1)
271-303: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Times(2)for three appended messages relies on same-commit-ts batching — assert it explicitly.Three messages (20, 10, 20) all resolve at watermark 20, so the expectation only holds if the flush path coalesces equal-commit-ts messages into one
DMLEvent. That implicit coupling makes the test fragile; consider asserting the row count per flushed event, and add a short scenario comment as the neighbouring tests do.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/kafka-consumer/writer_test.go` around lines 271 - 303, Strengthen TestWriterWrite_sortsOutOfOrderDMLByWatermark by recording each flushed DMLEvent’s row count, asserting the two events contain the expected counts for the 10 and coalesced 20 commit timestamps, and add a brief comment explaining that the two messages at commit timestamp 20 must be batched together.cmd/util/event_group.go (1)
61-104: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueAvoid the per-call
resolvedallocation by sorting in place.The compaction already partitions in place: resolved elements could be written into a second index of the same backing array, or better, do a stable partition and sort only the resolved segment. As written, every
ResolveIntocall allocates a slice sized to the whole buffer (cap 1024+ per table/partition), even in the common fully-ordered case.If you prefer the current shape, at least allocate lazily (
resolvedonly after the first resolvable message is seen) so no-op calls don't allocate.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/util/event_group.go` around lines 61 - 104, Update ResolveInto to avoid eagerly allocating the resolved slice for every call. Prefer compacting resolvable messages within g.messages’ existing backing array, then sort only the resolved segment when outOfOrder is detected, while preserving remaining-message order and cleanup behavior; if retaining the current structure, create resolved lazily only after the first message at or below resolve is encountered.cmd/pulsar-consumer/writer_test.go (1)
271-298: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSame implicit same-commit-ts batching assumption as the Kafka test.
Three appended messages but
Times(2); the expectation depends on the flush path coalescing equal commit-ts rows.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/pulsar-consumer/writer_test.go` around lines 271 - 298, The test TestWriterWrite_sortsOutOfOrderDMLByWatermark appends three DML messages but expects the mock sink's AddDMLEvent to be called only twice, depending on an implicit batching behavior where messages with the same commit-ts are coalesced during the flush. Add a comment in the test to explicitly document that the expectation of Times(2) relies on the writer's flush path coalescing the two messages with commit-ts 20 into a single AddDMLEvent call, making the batching assumption clear to future readers.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@cmd/kafka-consumer/writer.go`:
- Around line 617-628: Define one consistent policy for fallback rows below the
partition watermark and apply it in the Kafka branch around
messageWithPartitionCheck and the Pulsar mirrored fallback branch: either
explicitly document and rely on idempotent sink application, or prevent
advancing the partition watermark until the fallback row is flushed. Update
cmd/kafka-consumer/writer.go lines 617-628 and cmd/pulsar-consumer/writer.go
lines 506-515 with the same policy, ensuring EventsGroup.ResolveInto cannot
leave these rows applied out of order.
---
Outside diff comments:
In `@cmd/storage-consumer/consumer.go`:
- Around line 297-318: Update the storage consumer’s DML handling around the
group.HighWatermark checks to append every message unconditionally, including
commitTs values below the high watermark, relying on the group’s
sort-before-flush behavior. Remove the now-unnecessary enableTableAcrossNodes
branch and parameter from the relevant function, or retain it only as a log
field, and eliminate the fallback-drop path.
---
Nitpick comments:
In `@cmd/kafka-consumer/writer_test.go`:
- Around line 271-303: Strengthen TestWriterWrite_sortsOutOfOrderDMLByWatermark
by recording each flushed DMLEvent’s row count, asserting the two events contain
the expected counts for the 10 and coalesced 20 commit timestamps, and add a
brief comment explaining that the two messages at commit timestamp 20 must be
batched together.
In `@cmd/kafka-consumer/writer.go`:
- Around line 619-643: Rate-limit or aggregate the fallback-row warnings in the
relevant writer flow so broker disruptions cannot emit one Warn log per row,
while preserving visibility per table or flush. Update the affected structured
fields to use zap.Uint64 for partitionWatermark and zap.Stringer for protocol,
following the logging guidance; keep the existing fallback behavior unchanged.
In `@cmd/pulsar-consumer/writer_test.go`:
- Around line 271-298: The test TestWriterWrite_sortsOutOfOrderDMLByWatermark
appends three DML messages but expects the mock sink's AddDMLEvent to be called
only twice, depending on an implicit batching behavior where messages with the
same commit-ts are coalesced during the flush. Add a comment in the test to
explicitly document that the expectation of Times(2) relies on the writer's
flush path coalescing the two messages with commit-ts 20 into a single
AddDMLEvent call, making the batching assumption clear to future readers.
In `@cmd/util/event_group.go`:
- Around line 61-104: Update ResolveInto to avoid eagerly allocating the
resolved slice for every call. Prefer compacting resolvable messages within
g.messages’ existing backing array, then sort only the resolved segment when
outOfOrder is detected, while preserving remaining-message order and cleanup
behavior; if retaining the current structure, create resolved lazily only after
the first message at or below resolve is encountered.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 7e35b46b-c5d5-487a-bcba-86dfdaa87076
📒 Files selected for processing (7)
cmd/kafka-consumer/writer.gocmd/kafka-consumer/writer_test.gocmd/pulsar-consumer/writer.gocmd/pulsar-consumer/writer_test.gocmd/storage-consumer/consumer.gocmd/util/event_group.gocmd/util/event_group_test.go
|
/test all |
ti-chi-bot
Bot
added
size/XXL
|
/test all |
ti-chi-bot
Bot
added
approved
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: 3AceShowHand, asddongmen The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
[LGTM Timeline notifier]Timeline:
|
wk989898
added
the
needs-cherry-pick-release-8.5
|
In response to a cherrypick label: new pull request created to branch |
|
/cherry-pick release-nextgen-202603 |
|
@wk989898: new pull request created to branch DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository. |
4 participants
What problem does this PR solve?
Issue Number: close #5713
What is changed and how it works?
The consumer no longer drops fallback DML during append if the current message commit-ts is greater than global watermark. It keeps the DML in EventsGroup, then sorts and flushes it when a DDL or watermark arrives.
In the BAB case, if the consumer receives B first and later receives replayed A B, A is no longer ignored because its commitTs is smaller than the high watermark. It is buffered, then sorted into A B before writing.
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note
Summary by CodeRabbit
Summary by CodeRabbit
Bug Fixes
Tests